import org.apache.flume.Context;
import org.apache.flume.conf.ConfigurationException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.*;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

//这个是编写SQLSourceHelper 用来记录当前flume 的条数和时间
public class SQLSourceHelper {
    private static final Logger LOG = LoggerFactory.getLogger(SQLSourceHelper.class);
    private int runQueryDelay; //两次查询的时间间隔
    private int startFrom;   //查询语句开始id
    private int currentindex;  //查询语句当前id ， 每次查询之前需要查询元数据表
    private int recordSize = 0; //这个是每次查询发挥的结果条数
    private String table; // 这个是要操作的表
    private String columsToSelect; // 用户传入的查询的列
    private String customQuery; // 用户传入的查询语句
    private String query; // 构建的查询语句
    private String defaultcharsetResultSet; // 编码集
    private String columnId;
    private Context context;  // 用来获取上下文的配置

    private static  final  int DEFAULT_QUERY_DELAY = 10000;
    private static final int DEFAULT_START_VALUE  = 0;
    private static final String DEFAULT_COLUMNS_SELECT = "*";
    private static final String DEFAULT_CHARSET_RESULTSET = "UTF-8";
    private static Connection conn = null;
    private static Connection conn1 = null;
    private static PreparedStatement ps = null;
    private static String connectionURL,connectionUserName,connectionPassword;
    private static String connectionURL1,connectionUserName1,connectionPassword1;


    static {
        Properties p = new Properties();
        try{
            //获取到文件的路径信息并加载配置文件信息
            p.load(SQLSourceHelper.class.getClassLoader().getResourceAsStream("mysql.properties"));
            connectionURL1 = p.getProperty("url");
            connectionUserName1 = p.getProperty("user");
            connectionPassword1 = p.getProperty("password");
            Class.forName(p.getProperty("driver"));
        }catch (Exception e){
            LOG.error(e.toString());
        }
    }

    private static Connection getConnection(String url,String user,String pw){
        try{
            Connection conn = DriverManager.getConnection(url,user,pw);
            if(conn ==null){
                throw new SQLException();
            }
            return conn;
        }catch (SQLException e){
            e.printStackTrace();
        }
        return null;
    }
//创建构造方法
    SQLSourceHelper(Context context){
        //初始化上下文
        this.context = context;
        this.columsToSelect = context.getString("columns.to.select",DEFAULT_COLUMNS_SELECT);
        this.runQueryDelay = context.getInteger("run.query.delay",DEFAULT_QUERY_DELAY);
        this.startFrom = context.getInteger("start.from",DEFAULT_START_VALUE);
        this.defaultcharsetResultSet = context.getString("default.charset.resultset",DEFAULT_CHARSET_RESULTSET);
        //配置无默认值参数获取flume 任务配置文件中的参数
        this.table = context.getString("table");
        this.customQuery = context.getString("custom.query");
        connectionURL = context.getString("connection.url");
        connectionUserName = context.getString("connection.user");
        connectionPassword = context.getString("connection.password");
        this.columnId=context.getString("incremental.column.name");
        conn = getConnection(connectionURL, connectionUserName, connectionPassword);
        conn1 = getConnection(connectionURL1, connectionUserName1, connectionPassword1);
        //校验相应的配置信息 如果没有默认值的参数 也没赋予值 就抛出异常
        try{
            checkMandatoryProperties();
        }catch (ConfigurationException e){
            e.printStackTrace();
        }

        currentindex = getStatusDBIndex(startFrom);
        query = bulidQuery();
    }
    //校验相应的配置信息（表， 查询语句 以及数据库连接的参数）并抛出异常
    private void checkMandatoryProperties() throws ConfigurationException{
        if(table ==null){
            throw  new ConfigurationException("property table not set");
        }
        if(connectionURL == null){
            throw new ConfigurationException("connection.url property not set");
        }
        if(connectionUserName == null){
            throw new ConfigurationException("connection.user property not set");
        }
        if(connectionPassword == null){
            throw new ConfigurationException("connection.password property not set");
        }
        if(columnId ==null){
            throw  new ConfigurationException("incremental.column.name property not set");
        }
    }
    //这个是构建sql 查询语句 并实现查询操作
    private String bulidQuery(){
        String sql = "";
        currentindex =getStatusDBIndex(startFrom);
        LOG.info(currentindex+"");
        if(customQuery == null){
            sql = "select "+ columsToSelect+" from "+table;
        }else{
            sql = customQuery;
        }
        StringBuilder execSql = new StringBuilder(sql);
        if(!sql.contains("where")){
            execSql.append(" where ");
            execSql.append(columnId).append(">").append(currentindex);
            return execSql.toString();
        }else {
            int length = execSql.toString().length();
            return execSql.toString().substring(0, length - String.valueOf(currentindex).length()) + currentindex;
        }
    }
//这个是指定查询语句
    List<List<Object>> executeQuery(){
        try {
            customQuery = bulidQuery();
            //这个是存放结果的集合
            List<List<Object>> results = new ArrayList<>();
            if(ps ==null){
                ps = conn.prepareStatement(customQuery);
            }
            ResultSet result = ps.executeQuery();
            while (result.next()){
                List<Object> row = new ArrayList<>();
                for(int i = 1; i <= result.getMetaData().getColumnCount();i++){
                    row.add(result.getObject(i));
                }
                results.add(row);
            }
            //这个是获取到总的数量的集合
            LOG.info("execSql:"+customQuery+"\nresultSize: "+results.size());
            return results;
        }catch (SQLException e){
            LOG.error(e.toString());
            conn = getConnection(connectionURL,connectionUserName,connectionPassword);
        }
        return null;
    }
//将结果集转化成字符串 每一条数据是一个List 集合， 将每个小的list 集合转化成为字符串
    List<String> getAllRows(List<List<Object>> queryResult){
        List<String> allRows = new ArrayList<>();
        if(queryResult ==null || queryResult.isEmpty()){
            return allRows;
        }
        StringBuilder row = new StringBuilder();
        for(List<Object> rawRow: queryResult){
            Object value = null;
            for(Object aRawRow:rawRow){
                value = aRawRow;
                if(value==null){
                    row.append(",");
                }else{
                    row.append(aRawRow.toString()).append(",");
                }
            }
            allRows.add(row.toString());
            row = new StringBuilder();
        }
        return allRows;
    }
    //更新ooffset 元数据状态， 每次返回结果集之后调用， 必须记录每次查询的offset 值， 为程序中断续跑数据时使用
    //用id 作为 offset
    //如果查询到有重复组件就进行更新操作！
    void updateOffset2DB(int size){
        String sql = "insert into flume_meta(source_tab,currentIndex) VALUES ('"+
                this.table+"','"+(recordSize+=size)+"') on DUPLICATE key update "+
                "source_table = values(source_tab), currentIndex=values(currentIndex)";
        LOG.info("updateStatus Sql"+sql);
        execSql(sql);
    }
//执行的sql 语句getStatusDBIndex
private void execSql(String sql){
  try{
      ps = conn1.prepareStatement(sql);
      LOG.info("exec::"+sql);
      ps.execute();
  }      catch (SQLException e){
      e.printStackTrace();
  }
}
    //获取到元数据表中的id 的offset 偏移量
    private Integer getStatusDBIndex(int startFrom){
        String dbIndex = queryOne("select currentIndex from flume_meta where source_tab = '"+table+"'");
        if(dbIndex != null){
            return Integer.parseInt(dbIndex);
        }
        return startFrom;
    }
    private String queryOne(String sql){
        ResultSet result = null;
        try {
            //这个是返回的偏移量的值
            ps = conn1.prepareStatement(sql);
            result = ps.executeQuery();
            while (result.next()){
                return result.getString(1);
            }
        } catch (SQLException throwables) {
            throwables.printStackTrace();
        }
        return null;
    }
    void close(){
        try{
            ps.close();
            conn.close();
            conn1.close();
        }catch (SQLException ex){
            ex.printStackTrace();
        }
    }
    int getCurrentindex(){
        return currentindex;
    }
    void setCurrentindex(int newValue){
        currentindex = newValue;
    }
    int getRunQueryDelay(){
        return runQueryDelay;
    }
    String getQuery(){
        return query;
    }
    String getConnectionURL(){
        return connectionURL;
    }
    private boolean isCustomQuerySet(){
        return (customQuery !=null);
    }
    Context getContext(){
        return context;
    }
    public String getConnectionUserName(){
        return connectionUserName;
    }
    public String getConnectionPassword(){
        return connectionPassword;
    }
    String getDefalutCharsetResultSet(){
        return defaultcharsetResultSet;
    }
}

